package extract

import (
	"context"
	"github.com/siddontang/go-log/log"
	"hermes/message"
	"hermes/mysqldb"
	"hermes/processer"
	"hermes/util"
	"os"
	"strings"
	//"encoding/json"
	"fmt"
	"github.com/siddontang/go-mysql/mysql"
	"github.com/siddontang/go-mysql/replication"
)

func GetLastMysqlLogPos() mysql.Position {
	//check log file exist
	logPosFilePath := util.GetAppConfig().DataConfig.LogPosFilePath

	logPosExists,err :=util.Exists(logPosFilePath)
	if  err == nil && logPosExists {
		logPosContent,_ := util.ReadString(logPosFilePath)
		kv := strings.Split(logPosContent, "|")
		logFileName := kv[0]
		logPosStr := kv[1]
		logPos,converError := util.StrToUint32(logPosStr)
		if converError ==nil {
			return mysql.Position{logFileName,logPos}
		}else{
			log.Error(converError)
			panic("GetLastMysqlLogPos Error By Log Pos File")
		}
	}else {
		return mysqldb.GetLastMysqlLogPosByDb()
	}
}


func SaveLastMysqlLogPos(pos mysql.Position)  {
	logPosFilePath := util.GetAppConfig().DataConfig.LogPosFilePath
	logPosFileDir:= util.GetAppConfig().DataConfig.Dir

	isDrExists,_ := util.Exists(logPosFileDir)
	if !isDrExists {
		util.Mkdir(logPosFileDir,os.FileMode(777))
	}

	var logContent string
	logContent = fmt.Sprintf("%s|%d",pos.Name,pos.Pos)
	if len(logContent)>0{
		util.WriteString(logPosFilePath,logContent)
	}
}

func Start() {
	slaveId,host,port,username,password  := mysqldb.GetMysqlBaseConfig()
	cfg := replication.BinlogSyncerConfig{
		ServerID: slaveId,
		Flavor:   "mysql",
		Host:     host,
		Port:     port,
		User:     username,
		Password: password,
	}

	syncer := replication.NewBinlogSyncer(cfg)
	//streamer, _ := syncer.StartSync(mysql.Position{Name: "mysql-bin.000714", Pos: 0})
	lastPos := GetLastMysqlLogPos()
	streamer, _ := syncer.StartSync(lastPos)
	for {
		ev, _ := streamer.GetEvent(context.Background())
		// Dump event

		if ev == nil {
			fmt.Println("ev is null")
			SaveLastMysqlLogPos(lastPos)

			//panic("ev is null,may be last binlog is delete")
			continue
		}

		var r bool
		var schemaUpdate bool
		var tp string
		var nextPos mysql.Position
		switch ev.Header.EventType {
		case replication.WRITE_ROWS_EVENTv0:
		case replication.WRITE_ROWS_EVENTv1:
		case replication.WRITE_ROWS_EVENTv2:
			r = true
			tp = "INSERT"
			break

		case replication.UPDATE_ROWS_EVENTv0:
		case replication.UPDATE_ROWS_EVENTv1:
		case replication.UPDATE_ROWS_EVENTv2:
			r = true
			tp = "UPDATE"
			break

		case replication.DELETE_ROWS_EVENTv0:
		case replication.DELETE_ROWS_EVENTv1:
		case replication.DELETE_ROWS_EVENTv2:
			r = true
			tp = "DELETE"
			break

		case replication.TABLE_MAP_EVENT:
			r = false
			schemaUpdate = true
			tp = "table map"
			break

		case replication.QUERY_EVENT:
			r = false
			schemaUpdate = false
			break

		case replication.XID_EVENT:
			//TODO transition if need handle
			r = false
			schemaUpdate = false
			break

		case replication.ROTATE_EVENT:
			//rotateEvent := ev.Event.(*replication.RotateEvent)
			//nextPos = mysql.Position{string(rotateEvent.NextLogName),uint32(rotateEvent.Position)}
			r = false
			schemaUpdate = false
			break

		default:
			//fmt.Println(ev.Header.EventType)
			r = false
		}

		//if len(nextPos.Name) == 0 {
		//
		//}

		nextPos = syncer.GetNextPosition()

		if schemaUpdate {
			tableMapEvent := ev.Event.(*replication.TableMapEvent)
			processer.TableEventHandle(tableMapEvent,nextPos)
			continue
		}

		if r {
			rowsEvent := ev.Event.(*replication.RowsEvent)
			processMessage:=message.ProcessMessageBuild(tp,rowsEvent,nextPos)
			processer.MessageHandle(processMessage)
			//rowsEvent.Dump(os.Stdout)
			//_, ok := metaCache[tableName]
			//if !ok {
			//	//没有该表格的元数据
			//	metaCache[tableName] = TableMeta{
			//		ColumnNames: GetTableNames(tableName),
			//	}
			//}
			//
			////从缓存取元数据
			//meta := metaCache[tableName]
			//
			//rowsLength := len(rowsEvent.Rows)
			//data := make([]map[string]interface{}, rowsLength)
			//
			////构造多个行
			//for rowIndex, colsArray := range rowsEvent.Rows {
			//
			//	//构造每行的列
			//	cols := make([]map[string]interface{}, len(colsArray))
			//	for colIndex, colValue := range colsArray {
			//		cols[colIndex] = map[string]interface{}{
			//			"name":   meta.ColumnNames[colIndex],
			//			"index":  colIndex,
			//			"before": "unknow",
			//			"after":  colValue,
			//		}
			//	}
			//
			//	data[rowIndex] = map[string]interface{}{
			//		"database": string(rowsEvent.Table.Schema),
			//		"type":     tp,
			//		"table":    tableName,
			//		"cols":     cols,
			//	}
			//}
			//
			//event := map[string]interface{}{
			//	"size":        rowsLength,
			//	"data":        data,
			//	"retry_count": 0,
			//}
			//json.Marshal(event)
			////fmt.Println(jsonString)
			//fmt.Println(tp, " ", tableName, " rows:", rowsLength)
		}

		SaveLastMysqlLogPos(nextPos)
	}
}


